package com.chukun.multi.masterslave.reusable;

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Future;

/**
 *  Master参与者抽象类
 * @param <T> 子任务的类型
 * @param <V> 子任务处理结果的类型
 * @param <R> 原始任务处理结果的类型
 */
public abstract class AbstractMaster<T,V,R> {

    protected volatile Set<? extends SlaveSpec<T,V>> slaves;
    // 子任务派发算法策略
    private volatile SubTaskDispatchStrategy<T,V> dispatchStrategy;

    public AbstractMaster(){}

    protected void init() {
        // 创建子线程
        slaves = createSlaves();
        // 初始化任务分发策略
        dispatchStrategy = newSubTaskDispatchStrategy();

        for (SlaveSpec<T, V> slave : slaves) {
            slave.init();
        }
    }

    /**
     * 对子类暴露的服务方法。该类的子类需要定义一个具体的服务方法
     * @param params
     * @return
     * @throws Exception
     */
    protected R service(Object ... params) throws Exception{
        final TaskDivideStrategy taskDivideStrategy = newTaskDivideStrategy(params);

        /**
         * 对原始任务进行分解，并将分解得来的子任务派发给slave参与者实例，
         * 使用taskDivideStrategy 策略对任务进行分解
         * 使用dispatchStrategy 策略对任务进行分发
         */
        Iterator<Future<V>> subResults = dispatchStrategy.dispatch(slaves,taskDivideStrategy);

        // 等待slave实例处理结束
        for (SlaveSpec<T, V> slave : slaves) {
            slave.shutdown();
        }

        // 合并子任务的处理结果
        R result = combineResult(subResults);
        return result;
    }

    /**
     * 留给子类实现，用于创建原始任务分解算法策略
     * @param params
     * @return
     */
    protected abstract TaskDivideStrategy<T> newTaskDivideStrategy(Object...params);

    /**
     *用于创建子任务的派发算法策略。 默认使用轮询算法 {@link RoundRobinSubTaskDispatchStrategy}
     * @return
     */
    protected SubTaskDispatchStrategy<T,V> newSubTaskDispatchStrategy() {
        return new RoundRobinSubTaskDispatchStrategy<>();
    }

    /**
     * 留给子类实现。用于创建slave的实例
     * @return
     */
    protected abstract Set<? extends SlaveSpec<T,V>> createSlaves();

    /**
     * 留给子类实现。 用于合并子任务的处理结果.
     * @param subResults
     * @return
     */
    protected abstract R combineResult(Iterator<Future<V>> subResults);
}
